package com.lzq.mq.nameserver.event.spi.listener;

import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.constants.NameServerConstants;
import com.lzq.mq.common.enums.NameServerEventCode;
import com.lzq.mq.common.enums.NameServerResponseCode;
import com.lzq.mq.common.event.Listener;
import com.lzq.mq.nameserver.common.CommonCache;
import com.lzq.mq.nameserver.event.model.StartReplicationEvent;
import com.lzq.mq.nameserver.utils.NameserverUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;


/**
 * 注册子节点监听器
 * 用于主节处理子节点发送过来的开始同步请求
 * 1.验证用户名密码
 * 2.保存子节点的信息到CommonCache的ReplicationChannelManager中 ip:port -> channel
 */
@Slf4j
public class StartReplicationListener implements Listener<StartReplicationEvent> {

    @Override
    public void onReceive(StartReplicationEvent event) throws Exception {
        boolean isVerify = NameserverUtils.isVerify(event.getUser(), event.getPassword());
        ChannelHandlerContext channelHandlerContext = event.getChannelHandlerContext();
        if (!isVerify) {
            TcpMsg tcpMsg = new TcpMsg(NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode(),
                    NameServerResponseCode.ERROR_USER_OR_PASSWORD.getDesc().getBytes());
            channelHandlerContext.writeAndFlush(tcpMsg);
            channelHandlerContext.close();
            throw new IllegalAccessException("error account to connected!");
        }

        InetSocketAddress inetSocketAddress = (InetSocketAddress) channelHandlerContext.channel().remoteAddress();
        event.setSlaveIp(inetSocketAddress.getHostString());
        event.setSlavePort(String.valueOf(inetSocketAddress.getPort()));

        String reqId = event.getSlaveIp() + ":" + event.getSlavePort();
        channelHandlerContext.channel().attr(AttributeKey.valueOf(NameServerConstants.NAMESERVER_CHANNEL_ID)).set(reqId);     // todo 验证一下
        CommonCache.getReplicationChannelManager().put(reqId, channelHandlerContext);

        // 回复ack
        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.MASTER_START_REPLICATION_ACK.getCode(),new byte[0]);
        channelHandlerContext.writeAndFlush(tcpMsg);

        log.info("成功注册下游节点:{}", reqId);
        // todo 在此处同步, 处于子线程, 在此处直接将数据同步给子节点, 无法保证顺序

        // 推荐的做法是, 把同步的数据添加到一条队列中, 专门有一个现成从队列中取数据同步到各个节点

    }

}

